// -*-c++-*-

#ifndef _XBASE_MEMORY_TABLE_HPP_
#define _XBASE_MEMORY_TABLE_HPP_

#include <map>
#include <set>
#include <list>
#include <queue>

#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/foreach.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/shared_mutex.hpp>
#include <boost/thread/locks.hpp>

#include <protocol.hpp>
#include <stringable.hpp>
#include <config.hpp>
#include <ref_count.hpp>

#include "tablet_server_configuration.hpp"
#include "sstable.hpp"
#include "row_block.hpp"

namespace xbase {

	struct minor_compact_scanner
	{
		typedef boost::shared_ptr<minor_compact_scanner> pointer;

		struct greater_compare {
			inline
			bool operator() (const kv_scanner::pointer &first,
					 const kv_scanner::pointer &second) const {
				key_less_than key_lt;
				// second.entry.key < first.entry.key => first > second
				return key_lt(*second->get().m_key,
					      *first->get().m_key);
			}
		};

		bool eos() const {
			return m_queue.empty();
		}

		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top()->get();
		}

		void next() {
			if(! m_queue.empty())
			{
				kv_scanner::pointer scanner = m_queue.top();
				m_queue.pop();
				scanner->next();
				const bool eos = scanner->eos();
				if(! eos)
				{
					m_queue.push(scanner);
				}
			}
		}

		void add(const kv_scanner::pointer &scanner) {
			const bool eos = scanner->eos();
			if(! eos)
			{
				m_queue.push(scanner);
			}
		}
		
	private:
		std::priority_queue<kv_scanner::pointer,
				    std::vector<kv_scanner::pointer>,
				    greater_compare> m_queue;
	};

//
//
// 次要紧缩的实现
// 
// 0 获取内存表的写锁，生成新的内存表替换当前的，然后释放锁。老的内存表
// 设为只读。这保证了在执行次要紧缩的时候可以继续读写数据。
// 
// 1. 获取日志线程的回滚锁，这也同时保证了不能有两个次要紧缩操作同时进
// 行
//
// 2. 获取日志线程的当前序列号S（然后将日志线程中的序列号加一）
//
// 3. 以S作为文件名后缀，命名临时SSTable文件
//
// 4. 写入时，根据每个记录的列族所属的组，找到该组的目录，并在其下面创
// 建临时SSTable，完成后保存到映射中。每次在映射中找不到组名时，就创建
// 文件，打开并加入进去。然后将记录写入到那个SSTable文件中。完成后关闭
// 所有文件，再将所有文件改名为正式的SSTable文件（去掉前缀即可）。
//
// 5. 加载这些SSTable文件，添加到子表所管理的SSTable中。
//
// 6. 删除旧的内存表（写锁）
//
// 7. 释放回滚锁
//

	class memory_table
		: public boost::enable_shared_from_this<memory_table>
	{
		// typedef std::map<Key, Value> kv_map;
		// typedef boost::shared_ptr<kv_map> kv_map_pointer;
		// typedef std::list<kv_map_pointer> kv_map_list;
	public:
		typedef boost::shared_ptr<memory_table> pointer;
		
		typedef boost::mutex mutex_type;
		typedef std::map<std::string, table_schema::family_info> family_map_type;

		struct entry_type
		{
			row_block::pointer m_row_block; // 重要：读取数据时，一定要把该指针拷贝出去，才能保证内存不释放！
			boost::shared_ptr<mutex_type> m_row_mutex;

			explicit
			entry_type(const row_block::pointer row_block)
				: m_row_block(row_block)
				, m_row_mutex(new mutex_type()) {}
			
			const row_block::string_type *row_name() const {
				return m_row_block->row_name();
			}

			mutex_type &row_mutex() {
				return *m_row_mutex;
			}
		};

		// typedef std::pair< boost::shared_ptr<row_block>,
		// 		   boost::shared_ptr<FastLock> > block_type;
		struct less_than_compare {
			bool operator()(const row_block::string_type *first,
					const row_block::string_type *second) const {
				return (*first) < (*second);
			}
		};
		typedef std::map<const row_block::string_type *,
				 entry_type,
				 less_than_compare> impl_map_type;
		typedef boost::shared_ptr<impl_map_type> impl_map_pointer;
		typedef std::list<impl_map_pointer> impl_map_list;

	public:
		memory_table(uint32_t p_threshold,
			     uint32_t p_interval_threshold,
			     const std::string &p_table_name,
			     const std::string &p_group_name,
			     const uint64_t &tablet_id,
			     const std::string &p_tablet_name,
			     const std::string &start_row,
			     const std::string &end_row,
			     const table_schema::group_info &group_info,
			     const family_map_type &family_list)
			: m_table_name(p_table_name)
			, m_group_name(p_group_name)
			, m_tablet_id(tablet_id)
			, m_tablet_name(p_tablet_name)
			, m_start_row(start_row)
			, m_end_row(end_row)
			, m_group_info(group_info)
			, m_family_list(family_list)
			, m_table(new impl_map_type())
			, m_threshold(p_threshold)
			, m_interval_threshold(p_interval_threshold) {
			atomic_set(&m_size, 0);
			atomic_set(&m_last_compact_time, cur_seconds());
		}
		
		// interface
		
		// int add (const Key &p_key,
		// 	 const Value &p_value,
		// 	 uint64_t p_timestamp);

		// 传入属于同一行的记录列表
		void add(row_block::pointer &data);

		// 注意：返回空表示不需要扫描内存表
		kv_scanner::pointer get_scanner (kv_scan_context::pointer);

		bool is_exceed_threshold () const {
			const std::size_t cur_size = atomic_read(&m_size);
			return cur_size >= m_threshold;
		}

		bool is_exceed_interval() const {
			std::size_t oldest_time = cur_seconds();
			oldest_time -= m_interval_threshold;
			const std::size_t val = atomic_read(&m_last_compact_time);
			return (val <= oldest_time);
		}

		bool empty() const {
			return atomic_read(&m_size) == 0;
		}

		std::size_t bytes() const {
			return atomic_read(&m_size);
		}

		// 在内存表太大的时候，返回ture
		// 太大的标准：当前内存表大小 大于 MAX（用户设置的最大大小的4倍，1G）
		bool too_huge() const {
			const std::size_t hard_upper_bound = 1024 * 1024 * 1024; // 1G
			const std::size_t upper_bound = (std::max)(m_threshold, hard_upper_bound);
			const std::size_t limit = (std::min)(m_threshold * 4, upper_bound);
			const std::size_t cur_size = this->bytes();
			return (cur_size >= limit);
		}

		// 判断时间，用于周期性检查是否需要紧缩
		bool need_minor_compact() const {
			return (! empty()) &&
				(is_exceed_threshold() || is_exceed_interval());
		}

		// 不检查时间，用于在写记录后即时判断是否需要紧缩
		bool is_need_minor_compact_fastly() const {
			return (is_exceed_threshold() && (! empty()));
		}

		// 若 key 之后写入过值为空的记录，就认为该 key 是被删除了的，而
		// 不管被删除之后有没有写入新的记录；
		//
		// 若 key 中的时间戳是 NO_TIMESTAMP ，则只看最新写入的该记录的
		// 值是否为空，若为空，则返回真。其意义是，看该行/列当前有没有
		// 被删除。
		//
		// 若内存表中没有该记录，则返回假。
		//
		// 只判断是否被删除，不管版本数及是否失效等，那是合并紧缩的责任
		// 
		//bool is_deleted (const Key &p_key);

		// // 将内存中的数据写入到以 filename 命名的文件中
		// // 这里的container中的SSTable必须是auto_ptr的指针，否则会出现严重错误！
		// void compact(std::list<std::pair<String, boost::shared_ptr<SSTable> > > &container,
		// 	     const std::string &tablet_dir,
		// 	     const std::string &filename,
		// 	     const Table &table);

		std::string to_string () const;

		sstable::pointer compact (const uint64_t p_seqnum);

		bool is_working_table(const impl_map_pointer &table) const {
			return m_table == table;
		}

		// 用于memory_table_scanner中对当前工作表加读锁
		boost::shared_mutex &working_table_lock() {
			return m_lock;
		}

		// 外部加此读锁，将阻止历史内存表删除，进行阻止了历史内
		// 存表生成的SSTable加入到组中；该锁不能持太久，否则会
		// 阻塞次要紧缩过程。提供该接口是为在获取组的Scanner时
		// 加读锁，以避免遗漏或重复内存表和SSTable（比如同时将
		// 历史内存表和其生成的SSTable加入到扫描对象中）。
		boost::shared_mutex &history_table_lock() {
			return m_history_lock;
		}

		// 将当前内存表冻结
		void freeze ();
		void melt ();
		minor_compact_scanner::pointer get_compact_scanner();

	protected:
		// 将指定内存表紧缩
		// void compact(std::list<std::pair<String, boost::shared_ptr<SSTable> > > &container,
		// 	     const impl_map &memtable,
		// 	     const std::string &tablet_dir,
		// 	     const std::string &filename,
		// 	     const Table &table);

		// bool is_deleted (impl_map_pointer p_table,
		// 		 const Key &p_key) const;

		// void update_size (const Key &p_key,
		// 		  const Value &p_value);

		void update_size(uint32_t size) {
			atomic_add(size, &m_size);
		}
		void reset_size(uint32_t size) {
			atomic_set(&m_size, size);
		}
		void reset_compact_time(uint32_t sec) {
			// 检测时间错误
			LOG_ENV(const uint32_t val = atomic_read(&m_last_compact_time));
			LOG_IF((sec < val),
			       WARN,
			       "memory_table::reset_compact_time new time " << sec
			       << " should be larger than current compact time " << val);
			
			atomic_set(&m_last_compact_time, sec);
		}

		// 将冻结的内存表紧缩到磁盘上，成为SSTable
		sstable::pointer compact (minor_compact_scanner::pointer scanner,
					  const uint64_t p_seqnum);
		// 将冻结的内存表删除
		//void remove (const impl_map_pointer &p);
		
	private:
		friend class memory_table_scanner;

		std::string m_table_name;
		std::string m_group_name;
		uint64_t m_tablet_id;
		std::string m_tablet_name;
		std::string m_start_row;
		std::string m_end_row;

		const table_schema::group_info &m_group_info; // 对组中的group_info的引用
		const family_map_type &m_family_list;

		impl_map_pointer m_table;
		mutable boost::shared_mutex m_lock; // 仅在需要新插入行的数据才需要锁定，若只是更新行数据指定，不会修改map，视为读操作，不需要读锁
		
		atomic_t m_size; // 使用原子变量，在更新时不需要加锁
		atomic_t m_last_compact_time; // 使用原子变量，在更新时不需要加锁

		// 正在紧缩的内存表，按时间顺序放（即总是向后放），即最
		// 后的总是最新的
		//
		// 注意！这里的先后顺序假设跟扫描器有关，所以更改的时候
		// 需要同时更改get_scanner函数
		impl_map_list m_history_table;
		impl_map_list m_compacting_history_table;
		mutable boost::shared_mutex m_history_lock; // 按定义顺序加这个锁和工作表的锁

		const std::size_t m_threshold;
		const std::size_t m_interval_threshold;
	};

} // namespace xbase

#endif	// _XBASE_MEMORY_TABLE_HPP_
